package net.bwie.jtp.dim.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;

import java.util.HashMap;

/**
 * HbaseDimMapFunction是一个RichMapFunction，用于处理维度数据并准备将其写入HBase。
 * 它接受一个HashMap作为输入参数，该HashMap包含了表名到行键列名的映射。
 * 主要功能是为每个输入的JSON对象添加HBase表名、列族名和行键列名，以便于后续处理。
 */
public class HbaseDimMapFunction extends RichMapFunction<String, String> {
    // dimMap存储了表名到行键列名的映射，用于后续处理。
    private HashMap<String, String> dimMap;

    /**
     * 构造函数，初始化dimMap。
     * @param dimMap 一个HashMap，映射表名到行键列名。
     */
    public HbaseDimMapFunction(HashMap<String, String> dimMap) {
        this.dimMap = dimMap;
    }

    /**
     * 在功能开启时调用，此处未实现具体操作。
     * @param parameters 配置参数。
     * @throws Exception 如果开启过程中出现错误。
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
    }

    /**
     * 主要处理函数，为每个输入的JSON对象添加HBase相关的字段。
     * @param value 输入的JSON字符串，包含原始数据。
     * @return 修改后的JSON字符串，包含了HBase表名、列族名和行键列名。
     * @throws Exception 如果处理过程中出现错误。
     */
    @Override
    public String map(String value) throws Exception {
        // 将输入的字符串解析为JSONObject。
        JSONObject jsonObject = JSON.parseObject(value);
        // 添加HBase表名，格式为"dim_"前缀加上原始表名。
        jsonObject.put("hbase_table_name","dim_"+jsonObject.getString("table_name"));
        // 添加列族名，固定为"info"。
        jsonObject.put("family_name","info");
        // 添加行键列名，通过表名在dimMap中查找对应的行键列名。
        jsonObject.put("row_key_column",dimMap.get(jsonObject.getString("table_name")));
        // 返回修改后的JSON对象的字符串表示。
        return jsonObject.toJSONString();
    }

    /**
     * 在功能关闭时调用，此处未实现具体操作。
     * @throws Exception 如果关闭过程中出现错误。
     */
    @Override
    public void close() throws Exception {
        super.close();
    }
}
